//package com.zhz.mall.portal.kakfa;
//
//import cn.hutool.core.lang.Assert;
//import cn.hutool.core.util.IdUtil;
//import com.alibaba.fastjson.JSON;
//import lombok.AllArgsConstructor;
//import lombok.Builder;
//import lombok.Data;
//import lombok.NoArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.producer.ProducerRecord;
//import org.apache.kafka.common.header.Header;
//import org.apache.kafka.common.header.internals.RecordHeader;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.support.KafkaHeaders;
//import org.springframework.kafka.support.SendResult;
//import org.springframework.lang.NonNull;
//import org.springframework.lang.Nullable;
//import org.springframework.stereotype.Component;
//import org.springframework.util.concurrent.ListenableFuture;
//import org.springframework.util.concurrent.ListenableFutureCallback;
//
//import java.util.Arrays;
//import java.util.List;
//
///**
// * KafkaProducer
// * <p>
// * kafka producer
// *
// * @author zhouhengzhe
// */
//@Slf4j
//@Component("portalKafkaProducer")
//public class KafkaProducer {
//
//    private static final String KAFKA_MESSAGE_ID_HEADER = KafkaHeaders.PREFIX + "messageId";
//    private static final String SEND_USER_CODE = "userCode";
//    private static final String SEND_USER_NICKNAME = "nickname";
//
//
//    @Autowired
//    private KafkaTemplate<String, String> kafkaTemplate;
//
//
//    /**
//     * 发布消息
//     */
//    public ListenableFuture<SendResult<String, String>> send(String topic, Object data) {
//        return send(topic, null, data);
//    }
//
//
//    /**
//     * 发布消息
//     */
//    public ListenableFuture<SendResult<String, String>> send(String topic, @Nullable String key, Object data) {
//        Assert.notBlank(topic);
//        Assert.notNull(data);
//
//        String value = data instanceof String ? (String) data : JSON.toJSONString(data);
//        Assert.notBlank(value);
//
//        String messageId = IdUtil.fastSimpleUUID();
//        KafkaSendLogger.UserInfo userInfo = new KafkaSendLogger.UserInfo("admin","admin");
//
//        Header messageIdHeader = new RecordHeader(KAFKA_MESSAGE_ID_HEADER, messageId.getBytes());
//        Header userCodeHeader = new RecordHeader(SEND_USER_CODE,userInfo.getUserCode().getBytes());
//        Header userNicknameHeader = new RecordHeader(SEND_USER_NICKNAME, userInfo.getUserNickName().getBytes());
//
//        List<Header> headers = Arrays.asList(messageIdHeader, userCodeHeader, userNicknameHeader);
//        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, null, key, value, headers);
//
//        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(producerRecord);
//        future.addCallback(new KafkaSendLogger(producerRecord, messageId, userInfo));
//        return future;
//    }
//
//
//    @AllArgsConstructor
//    private static class KafkaSendLogger implements ListenableFutureCallback<SendResult<String, String>> {
//
//        private final ProducerRecord<String, String> producerRecord;
//        private final String messageId;
//        private final UserInfo userInfo;
//
//
//        @Override
//        public void onSuccess(SendResult<String, String> result) {
//            log.info("{}: kafka 消息[messageId: {}]发送成功, topic: {}, key: {}, value: {}",
//                    JSON.toJSONString(userInfo), messageId, producerRecord.topic(), producerRecord.key(), producerRecord.value());
//        }
//
//        @Override
//        public void onFailure(@NonNull Throwable ex) {
//            log.error("{}: kafka 消息[messageId: {}]发送失败, topic: {}, key: {}, value: {}",
//                    JSON.toJSONString(userInfo), messageId, producerRecord.topic(), producerRecord.key(), producerRecord.value());
//        }
//
//        @Data
//        @Builder
//        @AllArgsConstructor
//        @NoArgsConstructor
//        private static class UserInfo {
//            private String userCode;
//            private String userNickName;
//
//        }
//    }
//}
